package hello.java.app2;

import com.google.protobuf.ByteString;
import com.google.protobuf.Empty;
import grpc_demo.SendDtmfReply;
import grpc_demo.SendFileAudioRequest;
import grpc_demo.SendSipAudioReply;
import grpc_demo.TransmitterGrpc;
import io.grpc.Channel;
import io.grpc.stub.StreamObserver;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.util.Iterator;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;

public class TestClient {
    private static final Logger logger = Logger.getLogger(TestClient.class.getName());

    private final TransmitterGrpc.TransmitterBlockingStub blockingStub;
    private final TransmitterGrpc.TransmitterStub stub;
    private final int ChunkSize = 1024 * 32;

    public TestClient(Channel channel) {
        blockingStub = TransmitterGrpc.newBlockingStub(channel);
        stub = TransmitterGrpc.newStub(channel);
    }

    public void sendSipAudio() {
        Empty request = Empty.newBuilder().build();
        // 调用sendSipAudio方法，返回一个流式响应对象
        Iterator<SendSipAudioReply> responseObserver = blockingStub.sendSipAudio(request);
        // 循环读取响应流中的数据
        while (responseObserver.hasNext()) {
            // 如果需要查看语音流，请取消注释下面的日志

            // // 获取当前的响应对象
            // SendSipAudioReply current = responseObserver.next();
            // // 打印响应对象的内容
            // logger.info("Sip audio reply: " + current.getNumber() + " / " +
            // current.getData());
        }
    }

    public void sendDtmf() {
        Empty request = Empty.newBuilder().build();
        // 调用sendSipAudio方法，返回一个流式响应对象
        Iterator<SendDtmfReply> responseObserver = blockingStub.sendDtmf(request);
        // 循环读取响应流中的数据
        while (responseObserver.hasNext()) {
            // 获取当前的响应对象
            SendDtmfReply current = responseObserver.next();
            // 打印响应对象的内容
            logger.log(Level.INFO, "Dtmf reply:: " + current.getNumber() + " / " + current.getDtmf());
        }
    }

    public void sendFileAudio(String number, String hangup) {
        try {
            sendAudioFileCore(number, hangup);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        logger.info("sendFileAudio end:" + number);
    }

    private void sendAudioFileCore(String number, String hangup) throws InterruptedException {
        final CountDownLatch finishLatch = new CountDownLatch(1);
        StreamObserver<Empty> responseObserver = new StreamObserver<Empty>() {
            @Override
            public void onNext(Empty empty) {
                logger.info("onNext");
            }

            @Override
            public void onError(Throwable t) {
                logger.info("onError");
                logger.info("t:" + t);
                finishLatch.countDown();
            }

            @Override
            public void onCompleted() {
                logger.info("onCompleted");
                finishLatch.countDown();
            }
        };

        StreamObserver<SendFileAudioRequest> requestObserver = stub.sendFileAudio(responseObserver);
        try {
            File audioFile = new File("iflytek-16k.wav");
            // File audioFile = new File("test.wav");
            byte[] buffer = new byte[ChunkSize];
            int index = 0;
            try (FileInputStream fileStream = new FileInputStream(audioFile)) {
                while (true) {
                    int numBytesRead = fileStream.read(buffer);
                    if (numBytesRead == -1) {
                        logger.info("read file end.");
                        break;
                    } else {
                        logger.info(index + "read file numBytesRead:" + numBytesRead);
                        SendFileAudioRequest request = SendFileAudioRequest.newBuilder()
                                .setNumber(number)
                                .setData(ByteString.copyFrom(buffer, 0, numBytesRead))
                                .setHangup(Boolean.parseBoolean(hangup))
                                // .setData(ByteString.copyFrom(new byte[] { 1, 2, 3, 4, 5 }, 0, 5))
                                .build();
                        requestObserver.onNext(request);
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        } catch (RuntimeException e) {
            // Cancel RPC
            requestObserver.onError(e);
            throw e;
        }
        // Mark the end of requests
        requestObserver.onCompleted();

        // Receiving happens asynchronously
        if (!finishLatch.await(1, TimeUnit.MINUTES)) {
            logger.info("send audio file can not finish within 1 minutes");
        }
    }
}
